package com.example.server;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.SneakyThrows;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

public class CDCServer {

    Logger logger = Logger.getLogger(getClass().getName());

    private long serverId;

    private String host;

    private int port;

    private String db;

    private String username;

    private String password;

    private int retrySec = 5;

    private BinaryLogClient binaryLogClient;

    private BinaryLogClient.EventListener eventListener = new BinaryLogClient.EventListener() {
        @Override
        public void onEvent(Event event) {
            if (event.getData() instanceof TableMapEventData) {
                TableMapEventData eventData = event.getData();
                long tableId = eventData.getTableId();
                String tableName = eventData.getTable();
                tableMap.putIfAbsent(tableId, tableName);
            } else {
                if (event.getData() instanceof UpdateRowsEventData) {
                    //更新
                    UpdateRowsEventData updateRowsEventData = event.getData();
                    String tableName = tableMap.get(updateRowsEventData.getTableId());
                    logger.warning("table: " + tableName + " update ...");
                }

                if (event.getData() instanceof WriteRowsEventData) {
                    //新增
                    WriteRowsEventData writeRowsEventData = event.getData();
                    String tableName = tableMap.get(writeRowsEventData.getTableId());
                    logger.warning("table: " + tableName + " add ..." + writeRowsEventData.toString());
                }

                if (event.getData() instanceof DeleteRowsEventData) {
                    //删除
                    DeleteRowsEventData deleteRowsEventData = event.getData();
                    String tableName = tableMap.get(deleteRowsEventData.getTableId());
                    logger.warning("table: " + tableName + " del ...");
                }
            }
        }
    };

    private BinaryLogClient.LifecycleListener lifecycleListener = new BinaryLogClient.LifecycleListener() {
        @Override
        public void onConnect(BinaryLogClient client) {
            logger.warning("host: " + host + " db: " + db + " 连接成功 ...");
        }

        @Override
        public void onCommunicationFailure(BinaryLogClient client, Exception ex) {

        }

        @Override
        public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {

        }

        @SneakyThrows
        @Override
        public void onDisconnect(BinaryLogClient client) {
            connected  = false;
            connected = connect();
            while (!connected){
                TimeUnit.SECONDS.sleep(retrySec);
                logger.warning("host: " + host + " db: " + db + " 已断开连接 ...");
            }
        }
    };

    private Thread thread;

    private Map<Long, String> tableMap = new ConcurrentHashMap<>();

    private volatile boolean connected = true;

    public void start() {
        this.thread = new Thread(new Runnable() {
            @Override
            public void run() {
                connected = connect();
            }
        });
        this.thread.start();
    }

    public boolean connect(){
        boolean flag = true;
        try {
            binaryLogClient = new BinaryLogClient(host, port, db, username, password);
            binaryLogClient.setServerId(serverId);
            binaryLogClient.registerEventListener(eventListener);
            binaryLogClient.registerLifecycleListener(lifecycleListener);
            binaryLogClient.connect();
        } catch (Exception e) {
            flag = false;
            logger.warning("host：" + host + "数据库连接失败: " + e.getMessage());
        } finally {
            if (!flag) {
                try {
                    binaryLogClient.disconnect();
                } catch (IOException e) {
                    logger.warning("binaryLogClient 关闭失败: " + e.getMessage());
                }
            }
        }
        return flag;
    }

    public void stop() {
        if (this.thread.isAlive()) {
            this.thread.interrupt();
        }
    }

    public boolean isConnected(){
        return this.connected;
    }

    ///////////////// set get//////////////////
    public long getServerId() {
        return serverId;
    }

    public void setServerId(long serverId) {
        this.serverId = serverId;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getDb() {
        return db;
    }

    public void setDb(String db) {
        this.db = db;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public BinaryLogClient getBinaryLogClient() {
        return binaryLogClient;
    }

    public void setBinaryLogClient(BinaryLogClient binaryLogClient) {
        this.binaryLogClient = binaryLogClient;
    }

    public BinaryLogClient.EventListener getEventListener() {
        return eventListener;
    }

    public void setEventListener(BinaryLogClient.EventListener eventListener) {
        this.eventListener = eventListener;
    }

    public BinaryLogClient.LifecycleListener getLifecycleListener() {
        return lifecycleListener;
    }

    public void setLifecycleListener(BinaryLogClient.LifecycleListener lifecycleListener) {
        this.lifecycleListener = lifecycleListener;
    }

    public Thread getThread() {
        return thread;
    }

    public void setThread(Thread thread) {
        this.thread = thread;
    }

    public Map<Long, String> getTableMap() {
        return tableMap;
    }

    public void setTableMap(Map<Long, String> tableMap) {
        this.tableMap = tableMap;
    }

    public void setConnected(boolean connected) {
        this.connected = connected;
    }

    public CDCServer(long serverId, String host, int port, String db, String username, String password) {
        this.serverId = serverId;
        this.host = host;
        this.port = port;
        this.db = db;
        this.username = username;
        this.password = password;
    }

    public CDCServer(long serverId, String host, int port, String db, String username, String password, BinaryLogClient.EventListener eventListener, BinaryLogClient.LifecycleListener lifecycleListener) {
        this.serverId = serverId;
        this.host = host;
        this.port = port;
        this.db = db;
        this.username = username;
        this.password = password;
        this.eventListener = eventListener;
        this.lifecycleListener = lifecycleListener;
    }
}
